package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"bytes"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	"../labgob"
	"../labrpc"
)

// import "bytes"
// import "../labgob"

//const value
const (
	ELECTIONTIMEOUT  = 350
	HEARTBEATTIMEOUT = 100
	FOLLOWER         = 0
	CANDIDATE        = 1
	LEADER           = 2
)

//ApplyMsg struct
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in Lab 3 you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh; at that point you can add fields to
// ApplyMsg, but set CommandValid to false for these other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	IsSnapshot    bool
	Snapshots     []byte
	RaftStateSize int
}

//Entry struct
type Entry struct {
	Command interface{}
	Term    int
}

//Raft struct
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

	currentTerm int
	votedFor    int
	log         []Entry

	commitIndex int
	lastApplied int

	nextindex  []int
	matchindex []int

	//extra data
	state     int
	votes     int //votes in election
	prevtime  time.Time
	lastIndex int //由于Entry里没有加入index，因此加入此index重构
	lastTerm  int

	//resetElectionTimer chan struct{}
	applymsgCh     chan ApplyMsg
	applyRoutineCh chan struct{}
}

//GetState :
// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	var term int
	var isleader bool
	// Your code here (2A).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	term = rf.currentTerm
	if rf.state == LEADER {
		isleader = true
	} else {
		isleader = false
	}
	return term, isleader

}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.currentTerm)
	e.Encode(rf.votedFor)
	e.Encode(rf.log)
	e.Encode(rf.lastIndex)
	e.Encode(rf.lastTerm)
	data := w.Bytes()
	rf.persister.SaveRaftState(data)
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }

	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var x int
	var y int
	var z []Entry
	var lastindex int
	var lastterm int
	if d.Decode(&x) != nil ||
		d.Decode(&y) != nil ||
		d.Decode(&z) != nil ||
		d.Decode(&lastindex) != nil ||
		d.Decode(&lastterm) != nil {
		DPrintf("[%v]: error!", rf.me)
	} else {
		rf.currentTerm = x
		rf.votedFor = y
		rf.log = z
		rf.lastIndex = lastindex
		rf.lastTerm = lastterm
	}
}

//RequestVoteArgs struct
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateID  int
	LastLogIndex int
	LastLogTerm  int
}

//RequestVoteReply struct
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

//AppendEntriesArgs struct
type AppendEntriesArgs struct {
	Term         int
	LeaderID     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []Entry
	LeaderCommit int
}

//AppendEntriesReply struct
type AppendEntriesReply struct {
	Term          int
	ConflictIndex int
	Success       bool
}

//RequestVote :
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	//DPrintf("[%v]: above requestvotech", rf.me)
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if args.Term > rf.currentTerm {
		DPrintf("[%v]: found new term", rf.me)
		rf.state = FOLLOWER
		rf.votedFor = -1
		rf.currentTerm = args.Term
		rf.prevtime = time.Now()
		rf.persist()
		//rf.resetElectionTimer <- struct{}{}
	}
	reply.Term = rf.currentTerm
	if args.Term < rf.currentTerm {
		reply.VoteGranted = false
		return
	}
	//5.4的投票限制
	uptodate := false
	if args.LastLogTerm > rf.log[len(rf.log)-1].Term {
		uptodate = true
	} else if args.LastLogTerm == rf.log[len(rf.log)-1].Term && args.LastLogIndex >= len(rf.log)-1+rf.lastIndex {
		uptodate = true
	}
	if (rf.votedFor == -1 || rf.votedFor == args.CandidateID) && uptodate {
		rf.votedFor = args.CandidateID
		reply.VoteGranted = true
		rf.prevtime = time.Now()
		//rf.resetElectionTimer <- struct{}{}
		rf.persist()
		DPrintf(" [%v]: voted to %v, myterm %v, voterterm %v!\n", rf.me, args.CandidateID, rf.currentTerm, args.Term)
	} else {
		reply.VoteGranted = false
		rf.prevtime = time.Now()
		//rf.resetElectionTimer <- struct{}{}
	}

}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

//AppendEntries :
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	if args.Term > rf.currentTerm {
		DPrintf("[%v]: found new term %v from AppendEntries, myterm %v, transer to follower", rf.me, args.Term, rf.currentTerm)
		rf.state = FOLLOWER
		rf.votedFor = -1
		rf.currentTerm = args.Term
		rf.persist()
	}
	reply.Term = rf.currentTerm
	//rf.resetElectionTimer <- struct{}{}
	rf.prevtime = time.Now()
	if args.Term < rf.currentTerm {
		reply.Success = false
		DPrintf("[%v]: recv AppendEntries!but you're not Leader. yourterm %v, myterm %v", rf.me, args.Term, rf.currentTerm)
		rf.mu.Unlock()
		return
	}
	rf.currentTerm = args.Term
	if args.PrevLogIndex-rf.lastIndex < 0 {
		reply.ConflictIndex = rf.lastIndex
		reply.Success = false
		rf.mu.Unlock()
		return
	} else if args.PrevLogIndex >= len(rf.log)+rf.lastIndex {
		DPrintf("[%v]: args.PrevLogIndex %v >= len(rf.log) %v", rf.me, args.PrevLogIndex, len(rf.log))
		reply.ConflictIndex = len(rf.log) + rf.lastIndex
		reply.Success = false
		rf.mu.Unlock()
		return
	} else if args.PrevLogIndex-rf.lastIndex >= 0 && rf.log[args.PrevLogIndex-rf.lastIndex].Term != args.PrevLogTerm {
		DPrintf("[%v]: sameindex diffrent term rf.log[args.PrevLogIndex].Term %v, args.PrevLogTerm %v", rf.me, rf.log[args.PrevLogIndex-rf.lastIndex].Term, args.PrevLogTerm)
		reply.Success = false
		//rf.log = rf.log[:args.PrevLogIndex] //删除PrevLogIndex之后的entry
		conflictindex := args.PrevLogIndex
		for conflictindex-rf.lastIndex > 0 && rf.log[conflictindex-rf.lastIndex].Term == rf.log[args.PrevLogIndex-rf.lastIndex].Term { //accelerated log backtracking optimization
			conflictindex--
		}
		reply.ConflictIndex = conflictindex + 1
		rf.log = rf.log[:(reply.ConflictIndex - rf.lastIndex)]
		rf.persist()
		rf.mu.Unlock()
		return
	}
	if args.Entries != nil {
		DPrintf("[%v]: append entries success!", rf.me)
		DPrintf("[%v]: args{PreLogIndex: %v, PrevLogTerm: %v, Term: %v}", rf.me, args.PrevLogIndex, args.PrevLogTerm, args.Term)
		reply.Success = true
		//fmt.Printf("[%v]: len(rf.log) %v, args.Entries %v, args.PrevLogIndex %v\n", rf.me, len(rf.log), args.Entries, args.PrevLogIndex)
		if args.PrevLogIndex < len(rf.log)-1+rf.lastIndex && rf.log[args.PrevLogIndex+1-rf.lastIndex].Term != args.Entries[0].Term {
			rf.log = rf.log[:(args.PrevLogIndex + 1 - rf.lastIndex)]
		}
		if len(rf.log)-1+rf.lastIndex-args.PrevLogIndex < len(args.Entries) { //Append any new entries not already in the log
			rf.log = append(rf.log, args.Entries[(len(rf.log)-1+rf.lastIndex-args.PrevLogIndex):]...) //That's IMPORTANT!
			DPrintf("[%v]:", rf.me)
			for i, entry := range rf.log {
				//if _, ok := entry.Command.(int); ok {
				DPrintf("[%v]: {index: %v, term: %v, cmd: %v}\n", rf.me, i+rf.lastIndex, entry.Term, entry.Command)
				//}
			}
		}
		rf.persist()
	} else {
		reply.Success = true
	}
	if args.LeaderCommit > rf.commitIndex {
		if args.LeaderCommit > len(rf.log)-1+rf.lastIndex {
			rf.commitIndex = len(rf.log) - 1 + rf.lastIndex
		} else {
			rf.commitIndex = args.LeaderCommit
		}
	}
	if rf.commitIndex > rf.lastApplied {
		rf.mu.Unlock() //prevent deadlock with applyroutine
		DPrintf("[%v]: before apply msg. LastApplied %v, commitindex %v", rf.me, rf.lastApplied, rf.commitIndex)
		rf.applyRoutineCh <- struct{}{}
		DPrintf("[%v]: after apply msg. LastApplied %v, commitindex %v", rf.me, rf.lastApplied, rf.commitIndex)
	} else {
		rf.mu.Unlock()
	}
}

//appendEntriestoPeers :
func (rf *Raft) appendEntriestoPeers(oneEntry *Entry) {
	args := AppendEntriesArgs{}
	rf.mu.Lock()
	defer rf.mu.Unlock()
	args.LeaderID = rf.me
	args.Term = rf.currentTerm
	args.LeaderCommit = rf.commitIndex
	for i := 0; i < len(rf.peers); i++ {
		itmp := i
		if itmp != rf.me {
			if rf.nextindex[itmp] <= rf.lastIndex {
				args.PrevLogIndex = rf.lastIndex
				args.PrevLogTerm = rf.lastTerm
				args.Entries = rf.log[1:]
			} else {
				args.PrevLogIndex = rf.nextindex[itmp] - 1
				args.PrevLogTerm = rf.log[rf.nextindex[itmp]-1-rf.lastIndex].Term
				args.Entries = rf.log[(rf.nextindex[itmp] - rf.lastIndex):]
			}
			go rf.appendEntriestoOnePeer(itmp, args)
		}
	}
}

//appendEntriestoOnePeer :
func (rf *Raft) appendEntriestoOnePeer(itmp int, args AppendEntriesArgs) {
	reply := AppendEntriesReply{}
	if !rf.sendAppendEntries(itmp, &args, &reply) {
		return
	}
	rf.mu.Lock()
	if rf.currentTerm != args.Term || rf.state != LEADER {
		DPrintf("[%v]: old reply", rf.me)
		rf.mu.Unlock()
		return
	}
	if reply.Term > rf.currentTerm {
		DPrintf("[%v]: found new term %v from appendEntriesReply, myterm %v", rf.me, reply.Term, rf.currentTerm)
		rf.state = FOLLOWER
		rf.votedFor = -1
		rf.currentTerm = reply.Term
		rf.persist()
		rf.mu.Unlock()
		return
	}
	if reply.Success {
		if args.PrevLogIndex+len(args.Entries)+1 > rf.nextindex[itmp] {
			rf.nextindex[itmp] = args.PrevLogIndex + len(args.Entries) + 1
		}
		rf.matchindex[itmp] = rf.nextindex[itmp] - 1
		rf.updateCommitIndex()
		if rf.commitIndex > rf.lastApplied {
			rf.mu.Unlock()
			DPrintf("[%v]: before apply msg. LastApplied %v, commitindex %v", rf.me, rf.lastApplied, rf.commitIndex)
			rf.applyRoutineCh <- struct{}{}
			DPrintf("[%v]: after apply msg. LastApplied %v, commitindex %v", rf.me, rf.lastApplied, rf.commitIndex)
		} else {
			rf.mu.Unlock()
		}
	} else { //reply false
		DPrintf("[%v]: append failed!", rf.me)
		args.LeaderID = rf.me
		args.Term = rf.currentTerm
		args.LeaderCommit = rf.commitIndex
		/*rf.nextindex[itmp]--
		if rf.nextindex[itmp] < 1 {
			rf.nextindex[itmp] = 1
		}
		args.PrevLogIndex = rf.nextindex[itmp] - 1*/
		rf.nextindex[itmp] = reply.ConflictIndex //accelerated log backtracking optimization
		if reply.ConflictIndex <= rf.lastIndex {
			rf.nextindex[itmp] = rf.lastIndex + 1
			installsnapshotargs := InstallSnapshotArgs{rf.currentTerm, rf.me, rf.lastIndex, rf.lastTerm, rf.persister.ReadSnapshot()}
			go rf.sendInstallSnapshotRPCandReply(itmp, installsnapshotargs)
			args.PrevLogIndex = rf.lastIndex
			args.PrevLogTerm = rf.lastTerm
			args.Entries = rf.log[1:]
		} else {
			args.PrevLogIndex = rf.nextindex[itmp] - 1
			args.PrevLogTerm = rf.log[args.PrevLogIndex-rf.lastIndex].Term
			args.Entries = rf.log[rf.nextindex[itmp]-rf.lastIndex:]
		}
		go rf.appendEntriestoOnePeer(itmp, args)
		rf.mu.Unlock()
	}
}

//updateCommitIndex
func (rf *Raft) updateCommitIndex() {
	counts := 1
	for N := len(rf.log) - 1; N+rf.lastIndex > rf.commitIndex && N >= 0; N-- {
		if rf.log[N].Term == rf.currentTerm {
			for i := 0; i < len(rf.peers); i++ {
				if rf.matchindex[i] >= N+rf.lastIndex {
					counts++
				}
				if counts > len(rf.peers)/2 {
					rf.commitIndex = N + rf.lastIndex
					DPrintf("[%v]: update commitIndex %v", rf.me, rf.commitIndex)
					return
				}
			}
		} else {
			return
		}
		counts = 1
	}
}

//sendHeartbeats
func (rf *Raft) sendHeartbeats() {
	for {
		rf.mu.Lock()
		if rf.state != LEADER {
			DPrintf("[%v]: stop sending heartbeats", rf.me)
			rf.mu.Unlock()
			return
		}
		args := AppendEntriesArgs{}
		args.LeaderID = rf.me
		args.Term = rf.currentTerm
		args.LeaderCommit = rf.commitIndex
		//rf.mu.Unlock()
		for i := 0; i < len(rf.peers); i++ {
			itmp := i
			if itmp != rf.me {
				//DPrintf("[%v]: state %v, Send heartbeat to %v term %v", rf.me, rf.state, itmp, rf.currentTerm)
				if rf.nextindex[itmp] <= rf.lastIndex {
					args.PrevLogIndex = rf.lastIndex
					args.PrevLogTerm = rf.lastTerm
					args.Entries = rf.log[1:]
				} else {
					args.PrevLogIndex = rf.nextindex[itmp] - 1
					args.PrevLogTerm = rf.log[rf.nextindex[itmp]-1-rf.lastIndex].Term
					args.Entries = rf.log[rf.nextindex[itmp]-rf.lastIndex:]
				}
				go rf.appendEntriestoOnePeer(itmp, args)
			}
		}
		rf.mu.Unlock()
		time.Sleep(HEARTBEATTIMEOUT * time.Millisecond)
	}
}

//electionTimer :	electionTimer with channel
// check weather leader is alive
// run by Follower or Candidate state
/*func (rf *Raft) electionTimer() {
	for {
		//ms := ELECTIONTIMEOUT + (rand.Int63() % 150)
		ms := ELECTIONTIMEOUT + rand.New(rand.NewSource(time.Now().UnixNano())).Int63()%100
		select {
		case <-rf.resetElectionTimer:
			//DPrintf(" [%v]: He's alive!", rf.me)
		case <-time.After(time.Duration(ms) * time.Millisecond): //Timeout! Start election
			//DPrintf("[%v]: Time out! %v", rf.me, ms)
			rf.mu.Lock()
			if rf.state != LEADER {
				rf.state = CANDIDATE
				go rf.startelection()
			}
			rf.mu.Unlock()
			//return
		}
	}

}*/

//electionTimer : electionTimer without channel
func (rf *Raft) electionTimeout() {
	for {
		ms := ELECTIONTIMEOUT + rand.New(rand.NewSource(time.Now().UnixNano())).Int63()%150
		time.Sleep(150 * time.Millisecond)
		rf.mu.Lock()
		timeout := rf.prevtime.Add(time.Duration(ms) * time.Millisecond).Before(time.Now())
		if rf.state != LEADER && timeout {
			rf.state = CANDIDATE
			rf.startelection()
		}
		rf.mu.Unlock()

	}
}

//startelection :
func (rf *Raft) startelection() {
	DPrintf("[%v]: start election, mynewterm %v", rf.me, rf.currentTerm)
	rf.votes = 1
	rf.votedFor = rf.me
	rf.prevtime = time.Now()
	//rf.resetElectionTimer <- struct{}{}
	rf.currentTerm++
	args := RequestVoteArgs{}
	args.Term = rf.currentTerm // + 1
	args.CandidateID = rf.me
	if len(rf.log) == 0 {
		args.LastLogIndex = rf.lastIndex
		args.LastLogTerm = rf.lastTerm
	} else {
		args.LastLogIndex = len(rf.log) - 1 + rf.lastIndex
		args.LastLogTerm = rf.log[len(rf.log)-1].Term
	}
	rf.persist()
	for i := 0; i < len(rf.peers); i++ {
		itmp := i
		if itmp != rf.me {
			go rf.sendRequestVoteandcounts(itmp, &args)
		}
	}
}

//
func (rf *Raft) sendRequestVoteandcounts(server int, args *RequestVoteArgs) {
	//rf.mu.Lock()				//avoid lock during rpc
	//defer rf.mu.Unlock()
	reply := RequestVoteReply{}
	if !rf.sendRequestVote(server, args, &reply) {
		return
	}
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if rf.currentTerm != args.Term {
		DPrintf("[%v]: old reply", rf.me)
		return
	}
	if reply.Term > rf.currentTerm {
		DPrintf("[%v]: found new term", rf.me)
		rf.state = FOLLOWER
		rf.votedFor = -1
		rf.currentTerm = reply.Term
		rf.persist()
		return
	}
	if reply.VoteGranted {
		rf.votes++
		DPrintf("[%v]: votes: %v\n", rf.me, rf.votes)
		if rf.state == CANDIDATE && rf.votes > len(rf.peers)/2 {
			DPrintf("[%v]: I am leader! Current Term %v", rf.me, rf.currentTerm)
			//rf.currentTerm = rf.currentTerm + 1
			rf.votes = 0 //avoid repeat success election
			rf.state = LEADER
			for i := 0; i < len(rf.peers); i++ {
				rf.nextindex[i] = len(rf.log) + rf.lastIndex
				rf.matchindex[i] = -1
			}
			rf.persist()
			go rf.sendHeartbeats()
			return
		}
	}
}

//applyMsgToService
func (rf *Raft) applyMsgToService(index int) {
	applymsg := ApplyMsg{}
	applymsg.CommandValid = true
	applymsg.Command = rf.log[index-rf.lastIndex].Command
	applymsg.CommandIndex = index
	applymsg.RaftStateSize = rf.persister.RaftStateSize()
	rf.applymsgCh <- applymsg
	//rf.lastApplied = index
	DPrintf("[%v]: Apply msg success! {index: %v, command: %v}\n", rf.me, index, rf.log[index-rf.lastIndex].Command)
}

//applyRountine
func (rf *Raft) applyRountine() {
	for {
		<-rf.applyRoutineCh //block until some routine notice commitIndex > LastApplied
		rf.mu.Lock()

		if rf.commitIndex > rf.lastApplied {
			for i := rf.lastApplied + 1; i <= rf.commitIndex; i++ {
				if i-rf.lastIndex < 0 {
					break
				}
				rf.applyMsgToService(i)
			}
			rf.lastApplied = rf.commitIndex

			//check snapshot
			if rf.state == LEADER {
				applymsg := ApplyMsg{}
				applymsg.CommandValid = false
				applymsg.CommandIndex = rf.commitIndex
				applymsg.IsSnapshot = false
				applymsg.RaftStateSize = rf.persister.RaftStateSize()
				DPrintf("rf[%v]: before check snapshot", rf.me)
				rf.applymsgCh <- applymsg
				DPrintf("rf[%v]: wait check snapshot", rf.me)
				//wait for server's answer
				newapplymsg := <-rf.applymsgCh
				DPrintf("rf[%v]: after check snapshot", rf.me)
				if newapplymsg.CommandValid {
					//create snapshot
					DPrintf("rf[%v]: creating snapshot", rf.me)
					rf.CreateSnapshot(newapplymsg.Snapshots)
				}
			}
		}
		rf.mu.Unlock()
	}
}

//Start :
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := false

	// Your code here (2B).
	rf.mu.Lock()
	if rf.state != LEADER {
		rf.mu.Unlock()
		return index, term, isLeader
	}
	isLeader = true
	oneEntry := Entry{command, rf.currentTerm}
	rf.log = append(rf.log, oneEntry)
	DPrintf("[%v]:", rf.me)
	for i, entry := range rf.log {
		DPrintf("[%v]: {index: %v, term: %v, cmd: %v}\n", rf.me, i+rf.lastIndex, entry.Term, entry.Command)
	}
	index = len(rf.log) - 1 + rf.lastIndex
	term = rf.currentTerm
	rf.persist()
	rf.mu.Unlock()
	rf.appendEntriestoPeers(&oneEntry)

	return index, term, isLeader
}

//CreateSnapshot :
//Called By kvserver leader
func (rf *Raft) CreateSnapshot(snapshot []byte /*database map[string]string , index int*/) bool {
	//rf.mu.Lock()
	//defer rf.mu.Unlock()
	if rf.state != LEADER {
		return false
	}

	/*r := bytes.NewBuffer(snapshot)
	d := labgob.NewDecoder(r)
	var database map[string]string
	var reqhistory map[int64]int
	d.Decode(&database)
	d.Decode(&reqhistory)
	DPrintf("rf[%v]: database %v", rf.me, database)

	writesnapshot := new(bytes.Buffer)
	enc := labgob.NewEncoder(writesnapshot)
	enc.Encode(database)
	enc.Encode(reqhistory)
	//enc.Encode(rf.commitIndex)
	//enc.Encode(rf.log[rf.commitIndex-rf.lastIndex].Term)
	snapshotbyte := writesnapshot.Bytes()*/

	args := InstallSnapshotArgs{}
	args.LeaderID = rf.me
	args.Term = rf.currentTerm
	args.LastIncludedIndex = rf.commitIndex
	args.LastIncludedTerm = rf.log[rf.commitIndex-rf.lastIndex].Term
	args.Data = snapshot

	rf.log = rf.log[rf.commitIndex-rf.lastIndex+1:]
	rf.log = append([]Entry{{"", args.LastIncludedTerm}}, rf.log...)

	writestate := new(bytes.Buffer)
	e := labgob.NewEncoder(writestate)
	e.Encode(rf.currentTerm)
	e.Encode(rf.votedFor)
	e.Encode(rf.log)
	e.Encode(args.LastIncludedIndex)
	e.Encode(args.LastIncludedTerm)
	state := writestate.Bytes()

	rf.persister.SaveStateAndSnapshot(state, snapshot)
	//call installsnapshotrpc to each follower

	for i := 0; i < len(rf.peers); i++ {
		itmp := i
		if itmp != rf.me {
			go rf.sendInstallSnapshotRPCandReply(itmp, args)
		}
	}

	rf.lastIndex = args.LastIncludedIndex
	rf.lastTerm = args.LastIncludedTerm
	rf.lastApplied = args.LastIncludedIndex

	return true
}

//InstallSnapshotArgs :
type InstallSnapshotArgs struct {
	Term              int
	LeaderID          int
	LastIncludedIndex int
	LastIncludedTerm  int
	Data              []byte
}

//InstallSnapshotReply :
type InstallSnapshotReply struct {
	Term int
}

func (rf *Raft) sendInstallSnapshot(server int, args *InstallSnapshotArgs, reply *InstallSnapshotReply) bool {
	ok := rf.peers[server].Call("Raft.InstallSnapshot", args, reply)
	return ok
}

func (rf *Raft) sendInstallSnapshotRPCandReply(itmp int, args InstallSnapshotArgs) {
	reply := InstallSnapshotReply{}
	if !rf.sendInstallSnapshot(itmp, &args, &reply) {
		return
	}
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.currentTerm != args.Term || rf.state != LEADER {
		DPrintf("[%v]: old reply", rf.me)
		return
	}
	if reply.Term > rf.currentTerm {
		DPrintf("[%v]: found new term %v from installsnapshot, myterm %v", rf.me, reply.Term, rf.currentTerm)
		rf.state = FOLLOWER
		rf.votedFor = -1
		rf.currentTerm = reply.Term
		rf.persist()
		return
	}

}

//InstallSnapshot :
func (rf *Raft) InstallSnapshot(args *InstallSnapshotArgs, reply *InstallSnapshotReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	reply.Term = rf.currentTerm
	if args.Term < rf.currentTerm {
		DPrintf("[%v]: recv from Snapshot!but you're not Leader. yourterm %v, myterm %v", rf.me, args.Term, rf.currentTerm)
		return
	}

	if args.LastIncludedIndex < rf.lastIndex {
		DPrintf("[%v]: recv old snapshot", rf.me)
		return
	}
	//todo: receiver implementation
	if rf.lastIndex+len(rf.log)-1 <= args.LastIncludedIndex {
		rf.log = rf.log[0:0]
		rf.log = append(rf.log, Entry{"", args.LastIncludedTerm})
	} else {
		rf.log = rf.log[(args.LastIncludedIndex - rf.lastIndex + 1):]
		rf.log = append([]Entry{{"", args.LastIncludedTerm}}, rf.log...)
	}

	writestate := new(bytes.Buffer)
	e := labgob.NewEncoder(writestate)
	e.Encode(rf.currentTerm)
	e.Encode(rf.votedFor)
	e.Encode(rf.log)
	e.Encode(args.LastIncludedIndex)
	e.Encode(args.LastIncludedTerm)
	state := writestate.Bytes()
	rf.persister.SaveStateAndSnapshot(state, args.Data)

	rf.lastIndex = args.LastIncludedIndex
	rf.lastTerm = args.LastIncludedTerm
	rf.lastApplied = args.LastIncludedIndex

	//send snapshots to kvserver
	applymsg := ApplyMsg{}
	applymsg.CommandValid = false
	applymsg.IsSnapshot = true
	applymsg.Snapshots = args.Data
	DPrintf("[%v]: before send snapshot, lastindex %v\n", rf.me, rf.lastIndex)
	rf.applymsgCh <- applymsg
	DPrintf("[%v]: after send snapshot, lastindex %v\n", rf.me, rf.lastIndex)
}

//Kill :
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

//Make :
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).
	rf.currentTerm = 0
	rf.votedFor = -1
	//rf.resetElectionTimer = make(chan struct{})
	rf.prevtime = time.Now()
	rf.applymsgCh = applyCh
	rf.applyRoutineCh = make(chan struct{})
	rf.state = FOLLOWER
	rf.commitIndex = 0
	//rf.lastApplied = 0
	rf.nextindex = make([]int, 10)
	rf.matchindex = make([]int, 10)
	rf.lastIndex = 0
	rf.lastTerm = 0

	// initialize from state persisted befo re a crash
	rf.readPersist(persister.ReadRaftState())
	rf.lastApplied = rf.lastIndex
	if len(rf.log) == 0 {
		initentry := Entry{rf.lastIndex, rf.lastTerm}
		rf.log = append(rf.log, initentry)
	}

	//go rf.electionTimer()
	go rf.electionTimeout()
	go rf.applyRountine()

	return rf
}
